package com.deep.flink.oceanbase;

import com.ververica.cdc.connectors.oceanbase.source.RowDataOceanBaseDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import com.ververica.cdc.connectors.oceanbase.OceanBaseSource;
import com.ververica.cdc.connectors.oceanbase.table.OceanBaseTableSourceFactory;
import com.ververica.cdc.connectors.oceanbase.table.StartupMode;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

public class OceanBaseSourceExample {

    public static void main(String[] args) throws Exception {
        SourceFunction<String> oceanBaseSource =
                OceanBaseSource.<String>builder()
                        .rsList("127.0.0.1:2882:2881")  // set root server list
                        .startupMode(StartupMode.INITIAL) // set startup mode
                        .username("user@test_tenant")  // set cluster username
                        .password("pswd")  // set cluster password
                        .tenantName("test_tenant")  // set captured tenant name, do not support regex
                        .databaseName("test_db")  // set captured database, support regex
                        .tableName("test_table")  // set captured table, support regex
                        .hostname("127.0.0.1")  // set hostname of OceanBase server or proxy
                        .port(2881)  // set the sql port for OceanBase server or proxy
                        .logProxyHost("127.0.0.1")  // set the hostname of log proxy
                        .logProxyPort(2983)  // set the port of log proxy
//                        .deserializer(new JsonDebeziumDeserializationSchema())  // converts SourceRecord to JSON String
//                        .deserializer(new RowDataOceanBaseDeserializationSchema())  // converts SourceRecord to JSON String
                        .deserializer(new OceanBaseDeserializationSchemaImpl())  // converts SourceRecord to JSON String
                        .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // enable checkpoint
        env.enableCheckpointing(3000);

        env.addSource(oceanBaseSource).print().setParallelism(1);

        env.execute("Print OceanBase Snapshot + Commit Log");
    }
}
